package SparkStreaming;

import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function0;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.util.LongAccumulator;
import org.spark_project.guava.io.Files;
import scala.Tuple2;

import java.io.File;
import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;


/**
 * @program: MySpark
 * @description
 * @author: tkk fendoukaoziji
 * @create: 2019-04-24 09:47
 **/

/**
 * 通过单例创建一个广播变量
 */

class JavaWordBlackList{
    private static volatile Broadcast<List<String>> instance=null;
    public static Broadcast<List<String>> getInstance(JavaSparkContext jsc){
        if(instance==null){
            synchronized (JavaWordBlackList.class){
                if(instance==null){
                    List<String> wordBlackList = Arrays.asList("a", "b", "c");
                    instance = jsc.broadcast(wordBlackList);
                }
            }
        }
        return instance;
    }
}

class JavaDroppedWordsCount{
    private static volatile LongAccumulator instance=null;
    public static LongAccumulator getInstance(JavaSparkContext jsc){
        if(instance==null){
            synchronized (JavaDroppedWordsCount.class){
                if(instance==null){
                    instance=jsc.sc().longAccumulator("WordsInBlackListCounter");
                }
            }
        }
        return instance;
    }
}


public final class JavaRecoverableNetWordCount {
    private static final Pattern SPACE=Pattern.compile(" ");
    private static JavaStreamingContext createContext(String ip ,int port,String checkpointDirectory,String outputPath){
        System.out.println("Creating new Context");
        File outputFile=new File(outputPath);
        if(outputFile.exists()){
            outputFile.delete();
        }

        SparkConf sparkconf = new SparkConf().setAppName("JavaRecoverableNetWordCount").setMaster("local[*]");
        JavaStreamingContext ssc = new JavaStreamingContext(sparkconf, Durations.seconds(1));
        ssc.checkpoint(checkpointDirectory);
        JavaReceiverInputDStream<String> lines = ssc.socketTextStream(ip, port);
        JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(SPACE.split(x)).iterator());
        JavaPairDStream<String, Integer> wordCounts = words.mapToPair(s -> new Tuple2<>(s, 1)).reduceByKey((v1, v2) -> v1+v2);
        wordCounts.foreachRDD((rdd, time) ->{
            //获取或者创建广播变量
            Broadcast<List<String>> blackList = JavaWordBlackList.getInstance(new JavaSparkContext(rdd.context()));
            //获取或者创建累加器
            LongAccumulator dropWordCounter = JavaDroppedWordsCount.getInstance(new JavaSparkContext(rdd.context()));
            String counts = rdd.filter(wordCount -> {
                if (blackList.value().contains(wordCount._1())) {
                    dropWordCounter.add(wordCount._2);
                    return false;
                } else {
                    return true;
                }
            }).collect().toString();
            String output = "Counts at time " + time + " " + counts;
            System.out.println(output);
            System.out.println("Dropped " + dropWordCounter.value() + " word(s) totally");
            System.out.println("Appending to " + outputFile.getAbsolutePath());
            Files.append(output + "\n", outputFile, Charset.defaultCharset());
        } );
        return ssc;

    }

    public static void main(String[] args) throws InterruptedException {
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN); //设置日志级别warn
        String ip="node3";
        int port=1234;
        String chectPointDirectory= "text";
        String outputPath="outFile";
        Function0<JavaStreamingContext> createContextFunc = () -> createContext(ip, port, chectPointDirectory, outputPath);
        JavaStreamingContext ssc = JavaStreamingContext.getOrCreate(chectPointDirectory, createContextFunc);
        ssc.start();
        ssc.awaitTermination();
    }









}























